package org.yonggan.cmcc.config

import java.util

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import org.yonggan.cmcc.cache.JedisPoolUtil
import org.yonggan.cmcc.util.StringUtil

import scala.collection.mutable

/**
  * 提取kafka参数
  */
object KafkaParams {

  def kafkaOffset = {

    // 创建偏移量  goupid topic   partition offer
    val offset = new mutable.HashMap[TopicPartition, Long]()
    val jedis = JedisPoolUtil.getJedis
    // cmcc-offset  key val -->  topic-groupid parttion-offerset
    val offsetMap: util.Map[String, String] = jedis.hgetAll("cmcc-offset")
    // 将 java 集合转换成为 scala 集合
    import scala.collection.JavaConversions._
    for (map <- offsetMap) {
      val topicGroupPart = map._1.split("&")
      val partOffset = map._2
      //组件 offset
      offset += (new TopicPartition(topicGroupPart(0), topicGroupPart(2).toInt) -> partOffset.toLong)
    }
    offset
  }


  /**
    * 存储偏移量
    */
  def saveKafkaOffset(offsetArr: Array[OffsetRange]) = {
    // 存储偏移量
    val jedis = JedisPoolUtil.getJedis
    // 更新 偏移量
    offsetArr.foreach(f => {

      var key = StringUtil.append(f.topic, "each_stream", "&")
      key = StringUtil.append(key, f.partition.toString, "&")

      // 存储数据
      jedis.hincrBy("cmcc-offset", key, f.untilOffset)
      println(s"记录偏移量 -->  ${key} : ${f.untilOffset} ")
    })
    jedis.close()
  }

}
